KAFKA-12648: fix bug where thread is re-added to TopologyMetadata when shutting down by ableegoldman · Pull Request #11857 · apache/kafka

ableegoldman · 2022-03-07T09:14:44Z

We used to call TopologyMetadata#maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion when a thread was being unregistered/shutting down, to check if any of the futures listening for topology updates had been waiting on this thread and could be completed. Prior to invoking this we make sure to remove the current thread from the TopologyMetadata's threadVersions map, but this thread is actually then re-added in the #maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion call.

To fix this, we should break up this method into separate calls for each of its two distinct functions, updating the version and checking for topology update completion. When unregistering a thread, we should only invoke the latter method

ableegoldman · 2022-03-07T09:14:54Z

cc @wcarlson5

ableegoldman · 2022-03-07T10:56:51Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

-            log.info("StreamThread has detected an update to the topology, triggering a rebalance to refresh the assignment");
-            if (topologyMetadata.isEmpty()) {
-                mainConsumer.unsubscribe();
-            }
-            topologyMetadata.maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion(getName());


All of this has been moved into taskManager#handleTopologyUpdates

ableegoldman · 2022-03-07T10:59:21Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java

     */
    void handleTopologyUpdates() {
-        tasks.maybeCreateTasksFromNewTopologies();
+        final Set<String> currentNamedTopologies = topologyMetadata.updateThreadTopologyVersion(Thread.currentThread().getName());


This isn't the main fix, but we were playing a little fast and loose with the topology version we were reporting having ack'ed -- tightened this up by first atomically updating the topology version and saving the set of current named topologies, then doing the actual update handling, and then checking the listeners and completing any finished add/remove topology requests

ableegoldman · 2022-03-07T10:59:51Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java

        public ReentrantLock topologyLock = new ReentrantLock();
        public Condition topologyCV = topologyLock.newCondition();
-        public List<TopologyVersionWaiters> activeTopologyWaiters = new LinkedList<>();
+        public List<TopologyVersionListener> activeTopologyUpdateListeners = new LinkedList<>();


Just renamed from waiters to listeners

Also another quick question regarding why we need to keep topologyVersion an atomic long? Seems besides the getters all of its updators are under the lock as well.

Good find, yeah I believe it no longer needs to be an AtomicLong, I'll change back to long

Oh right actually no, we. do still need it to be an AtomicLong as we check it in the StreamThread main loop when looking for topology updates. And obviously we don't want to have to grab the full lock for that

ableegoldman · 2022-03-07T11:03:13Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java

-            final Iterator<TopologyVersionWaiters> iterator = version.activeTopologyWaiters.listIterator();
-            TopologyVersionWaiters topologyVersionWaiters;
+            version.topologyLock.lock();
            threadVersions.put(threadName, topologyVersion());


@wcarlson5 / @guozhangwang / @vvcephei This is the main fix -- we need to split out the version update where we add the current thread with the latest topology version to this threadVersions map, since this of course should only be done when we're reacting to a topology update.

The other function of the method was to check whether we could complete any of the queued listeners, which is why we were invoking this when shutting down a thread. Splitting this out into a separate method avoids ghost threads being left behind in the threadVersions map

ableegoldman · 2022-03-07T11:05:36Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java

-                    }
+                topologyVersionListener = iterator.next();
+                final long topologyVersionWaitersVersion = topologyVersionListener.topologyVersion;
+                if (minThreadVersion >= topologyVersionWaitersVersion) {


I also refactored this slightly to optimize/clean up this method. It's less about the optimization as we should generally not have too many threads per KafkaStreams runtime, but I found it much easier to follow the logic by computing the minimum version across all threads and then completing all futures listening for the topology to be updated up to that version

wcarlson5

Nice catch. LGTM I don't see anything that needs to get into this PR. I appreciate the test fix too

wcarlson5 · 2022-03-07T17:19:46Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java

-                    }
+                topologyVersionListener = iterator.next();
+                final long topologyVersionWaitersVersion = topologyVersionListener.topologyVersion;
+                if (minThreadVersion >= topologyVersionWaitersVersion) {


I think we also want to remove the listeners for threads that were removed as well right?

The listeners are for the caller threads not stream threads right? I thought since the thread is removed, it would not be counted in the getMinimumThreadVersion() and hence would not block the listeners to be removed.

No, the listeners are for stream threads. They get added in the task manager. Once all threads are at the version the future blocking the calling thread is completed.

Hmm... just to make sure we are talking about version.activeTopologyUpdateListeners right? These listeners are for the calling thread of the removeNamedTopology / addNamedTopology / start, which would get the wraped futures these listeners are constructed on.

Anyways, my understanding is that when a thread is removed, the getMinimumThreadVersion returned version would not take that removed thread into consideration, so that even the removed thread's version is low it would not block the future being completed.

Ah @guozhangwang yeah the getMinimumThreadVersion should take care of it.

Yeah I think this was already resolved but just to clarify for anyone else reading this/ourselves in the future, yes, the listeners are for the callers of add/removeNamedTopology 👍

guozhangwang

@ableegoldman the change lgtm overall. But I have a meta question about our synchronization: in the TopologyMetadata class we have two synchronization manners: 1) we use a lock for any changes to the TopologyVersion object, 2) we made the threadVersions a concurrent hashmap, but not all modifications (e.g. register/deregister thread) would require the lock in 1).

That means the TopologyVersion object may not be always consistent from the threadVersions map. Is this okay or intended by our design? If not, maybe we can just make the threadVersions be part of the TopologyVersion and be always updated with the same lock.

guozhangwang · 2022-03-07T18:10:38Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java

-                    }
+                topologyVersionListener = iterator.next();
+                final long topologyVersionWaitersVersion = topologyVersionListener.topologyVersion;
+                if (minThreadVersion >= topologyVersionWaitersVersion) {


The listeners are for the caller threads not stream threads right? I thought since the thread is removed, it would not be counted in the getMinimumThreadVersion() and hence would not block the listeners to be removed.

streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java

guozhangwang · 2022-03-07T18:36:26Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java

-                    }
+                topologyVersionListener = iterator.next();
+                final long topologyVersionWaitersVersion = topologyVersionListener.topologyVersion;
+                if (minThreadVersion >= topologyVersionWaitersVersion) {


Hmm... just to make sure we are talking about version.activeTopologyUpdateListeners right? These listeners are for the calling thread of the removeNamedTopology / addNamedTopology / start, which would get the wraped futures these listeners are constructed on.

Anyways, my understanding is that when a thread is removed, the getMinimumThreadVersion returned version would not take that removed thread into consideration, so that even the removed thread's version is low it would not block the future being completed.

ableegoldman · 2022-03-08T07:58:04Z

All test failures are unrelated, going to merge this now

…n shutting down (apache#11857) (#674) We used to call TopologyMetadata#maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion when a thread was being unregistered/shutting down, to check if any of the futures listening for topology updates had been waiting on this thread and could be completed. Prior to invoking this we make sure to remove the current thread from the TopologyMetadata's threadVersions map, but this thread is actually then re-added in the #maybeNotifyTopologyVersionWaitersAndUpdateThreadsTopologyVersion call. To fix this, we should break up this method into separate calls for each of its two distinct functions, updating the version and checking for topology update completion. When unregistering a thread, we should only invoke the latter method Reviewers: Guozhang Wang <guozhang@confluent.io>, Walker Carlson <wcarlson@confluent.io> Co-authored-by: A. Sophie Blee-Goldman <sophie@confluent.io>

fix thread version

6aa0a25

ableegoldman requested a review from guozhangwang March 7, 2022 09:14

ableegoldman added 3 commits March 7, 2022 02:04

return current named topologies within lock

844f582

notify listeners after unsubscribing

f555950

consolidate everything into TaskManager#handleUpdates

b6bbdbd

ableegoldman requested a review from vvcephei March 7, 2022 10:56

ableegoldman commented Mar 7, 2022

View reviewed changes

wcarlson5 approved these changes Mar 7, 2022

View reviewed changes

wcarlson5 reviewed Mar 7, 2022

View reviewed changes

guozhangwang reviewed Mar 7, 2022

View reviewed changes

ableegoldman merged commit fc7133d into apache:trunk Mar 8, 2022

Conversation

ableegoldman commented Mar 7, 2022

Uh oh!

ableegoldman commented Mar 7, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

wcarlson5 left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ableegoldman commented Mar 8, 2022

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Comments

wcarlson5 left a comment •

edited

Loading